package main

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"os"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/louloulin/dataflare/pkg/workflow"
)

func main() {
	// Check if MySQL DSN is provided
	dsn := os.Getenv("MYSQL_DSN")
	if dsn == "" {
		log.Fatal("MYSQL_DSN environment variable is required")
	}

	// Setup MySQL database
	if err := setupDatabase(dsn); err != nil {
		log.Fatalf("Failed to setup database: %v", err)
	}

	// Parse workflow definition
	definition, err := workflow.ParseFromFile("examples/mysql-workflow.yaml")
	if err != nil {
		log.Fatalf("Failed to parse workflow: %v", err)
	}

	fmt.Printf("Running workflow: %s\n", definition.Name)

	// Create engine
	engine, err := workflow.NewEngine()
	if err != nil {
		log.Fatalf("Failed to create engine: %v", err)
	}
	defer engine.Shutdown()

	// Deploy workflow
	ctx := context.Background()
	fmt.Println("Deploying workflow...")
	workflowID, err := engine.DeployWorkflow(ctx, definition)
	if err != nil {
		log.Fatalf("Failed to deploy workflow: %v", err)
	}
	fmt.Printf("Workflow deployed with ID: %s\n", workflowID)

	// Start workflow
	fmt.Println("Starting workflow...")
	runID, err := engine.StartWorkflow(ctx, workflowID, nil)
	if err != nil {
		log.Fatalf("Failed to start workflow: %v", err)
	}
	fmt.Printf("Workflow started with run ID: %s\n", runID)

	// Wait for workflow to complete
	fmt.Println("Waiting for workflow to complete...")
	for {
		status, err := engine.GetWorkflowStatus(ctx, runID)
		if err != nil {
			log.Fatalf("Failed to get workflow status: %v", err)
		}

		fmt.Printf("Workflow status: %s\n", status.Status)

		if status.Status == "completed" || status.Status == "failed" {
			break
		}

		time.Sleep(1 * time.Second)
	}

	// Verify results
	if err := verifyResults(dsn); err != nil {
		log.Fatalf("Failed to verify results: %v", err)
	}

	fmt.Println("Workflow completed successfully!")
}

func setupDatabase(dsn string) error {
	// Connect to MySQL
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return fmt.Errorf("failed to connect to MySQL: %w", err)
	}
	defer db.Close()

	// Create test database
	_, err = db.Exec("CREATE DATABASE IF NOT EXISTS dataflare_test")
	if err != nil {
		return fmt.Errorf("failed to create test database: %w", err)
	}

	// Use test database
	_, err = db.Exec("USE dataflare_test")
	if err != nil {
		return fmt.Errorf("failed to use test database: %w", err)
	}

	// Create source table
	_, err = db.Exec(`
		CREATE TABLE IF NOT EXISTS users (
			id INT PRIMARY KEY,
			name VARCHAR(100),
			email VARCHAR(100),
			age INT,
			created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
		)
	`)
	if err != nil {
		return fmt.Errorf("failed to create source table: %w", err)
	}

	// Create destination table
	_, err = db.Exec(`
		CREATE TABLE IF NOT EXISTS processed_users (
			id INT PRIMARY KEY,
			name VARCHAR(100),
			email VARCHAR(100),
			age INT,
			age_months INT,
			processed BOOLEAN,
			created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
		)
	`)
	if err != nil {
		return fmt.Errorf("failed to create destination table: %w", err)
	}

	// Insert test data
	_, err = db.Exec(`
		INSERT INTO users (id, name, email, age) VALUES
		(1, 'Alice', 'alice@example.com', 30),
		(2, 'Bob', 'bob@example.com', 25),
		(3, 'Charlie', 'charlie@example.com', 35)
		ON DUPLICATE KEY UPDATE
		name = VALUES(name),
		email = VALUES(email),
		age = VALUES(age)
	`)
	if err != nil {
		return fmt.Errorf("failed to insert test data: %w", err)
	}

	fmt.Println("Database setup completed successfully")
	return nil
}

func verifyResults(dsn string) error {
	// Connect to MySQL
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return fmt.Errorf("failed to connect to MySQL: %w", err)
	}
	defer db.Close()

	// Use test database
	_, err = db.Exec("USE dataflare_test")
	if err != nil {
		return fmt.Errorf("failed to use test database: %w", err)
	}

	// Query processed users
	rows, err := db.Query("SELECT id, name, email, age, age_months, processed FROM processed_users ORDER BY id")
	if err != nil {
		return fmt.Errorf("failed to query processed users: %w", err)
	}
	defer rows.Close()

	// Print results
	fmt.Println("\nProcessed Users:")
	fmt.Println("--------------------------------------------------")
	fmt.Printf("%-5s %-15s %-25s %-5s %-10s %-10s\n", "ID", "Name", "Email", "Age", "Age Months", "Processed")
	fmt.Println("--------------------------------------------------")

	count := 0
	for rows.Next() {
		var id, age, ageMonths int
		var name, email string
		var processed bool
		if err := rows.Scan(&id, &name, &email, &age, &ageMonths, &processed); err != nil {
			return fmt.Errorf("failed to scan row: %w", err)
		}
		fmt.Printf("%-5d %-15s %-25s %-5d %-10d %-10t\n", id, name, email, age, ageMonths, processed)
		count++
	}
	fmt.Println("--------------------------------------------------")
	fmt.Printf("Total: %d records\n", count)

	// Verify count
	if count != 3 {
		return fmt.Errorf("expected 3 processed records, got %d", count)
	}

	// Verify age_months calculation
	var count30 int
	err = db.QueryRow("SELECT COUNT(*) FROM processed_users WHERE age = 30 AND age_months = 360").Scan(&count30)
	if err != nil {
		return fmt.Errorf("failed to verify age_months calculation: %w", err)
	}
	if count30 != 1 {
		return fmt.Errorf("expected 1 record with age 30 and age_months 360, got %d", count30)
	}

	// Verify processed flag
	var countProcessed int
	err = db.QueryRow("SELECT COUNT(*) FROM processed_users WHERE processed = TRUE").Scan(&countProcessed)
	if err != nil {
		return fmt.Errorf("failed to verify processed flag: %w", err)
	}
	if countProcessed != 3 {
		return fmt.Errorf("expected 3 records with processed = TRUE, got %d", countProcessed)
	}

	return nil
}
